package com.实战;

import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class NewsServiceSubscriber implements Subscriber<String> {

	final Queue<String> mailbox = new ConcurrentLinkedQueue<>();
	final int take;
	final AtomicInteger remaining = new AtomicInteger();
	Subscription subscription;

	public NewsServiceSubscriber(int take) {
		this.take = take;
	}

	// 我们保存接收到的Subscription并将先前具有偏好的用户的新读取吞吐量发送到服务器
	@Override
	public void onSubscribe(Subscription s) {
		subscription = s;
		subscription.request(take);
	}

	// 处理新摘要的逻辑很直观，它只是将消息放入queue邮箱的过程
	@Override
	public void onNext(String news) {
		mailbox.offer(news);
	}

	@Override
	public void onError(Throwable t) {
		// TODO Auto-generated method stub

	}

	@Override
	public void onComplete() {
		// TODO Auto-generated method stub

	}

	// 首先，为了表明邮箱可能是空的，我们依赖于Optional，同时，第一步，我们尝试从邮箱中获取最新的未读新闻摘要。如果邮箱中没有未读新闻，我们返回Optional.<String>empty()
	// 在存在可用摘要的情况下，我们会减少计数器，该计数器表示先前已经从新闻服务请求的未读消息的数量。如果我们仍在等待某些消息，我们将返回已完成的Optional.<String>of(news)
	// 否则我们将另外请求新的摘要并重置剩余新消息的计数器

	// eventuallyReadDigest的实现只不过是真实用户的一些操作，例如打开邮箱、检查新消息、读取邮件并将其标记为已读，或者只是在没有新内容可以与之交互时关闭邮箱
	public Optional<String> eventuallyReadDigest() {
		String news = mailbox.poll();
		if(news != null) {
			if(remaining.decrementAndGet() == 0) {
				subscription.request(take);
				remaining.set(take);
			}
			return Optional.<String>of(news);
		}
		return Optional.<String>empty();
	}

}
